package com.qdu;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessor;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.coprocessor.RegionObserver;
import org.apache.hadoop.hbase.wal.WALEdit;

import java.io.IOException;
import java.net.URL;
import java.util.List;
import java.util.Optional;

/**
 * 表中插入数据之前，判断是否存在cf1:url列，存在的话，从中提取hostname，并存入cf1:hostname中
 * alter 'cls02', METHOD => 'table_att', 'coprocessor' => 'hdfs://niit-master:9000/user/niit/hbaselib/HBaseCoprocessor-1.0-SNAPSHOT.jar|com.qdu.UrlHostNameCoprocessor|1001|'
 */
public class UrlHostNameCoprocessor
        // 1. 实现接口
        implements RegionObserver, RegionCoprocessor {
    // 2. 重载2个接口中必要的方法，尤其是：prePut
    @Override
    public Optional<RegionObserver> getRegionObserver() {
        return Optional.of(this);
    }

    @Override
    public void prePut(
            ObserverContext<RegionCoprocessorEnvironment> c,
            Put put,
            WALEdit edit,
            Durability durability) throws IOException {
        // 2.1 检查put的内容是否符合要求
        List<Cell> cellList = put.get("cf1".getBytes(),"url".getBytes());
        // 检查cf1:url单元格是否存在
        // 防止操作别的列时，此协处理器继续执行后续的代码
        if(cellList==null || cellList.size()==0){
            return;
        }
        // 2.2 只需要更改put对象中的列对应的单元格即可，不需要执行put等动作命令对应的方法
        // http://www.qdu.edu.cn/news/1.html    ::: url
        // www.qdu.edu.cn           ::: hostname

        // 借助将数据转换为目标类型的对象是否成功，来变相的验证此数据是否格式正确
        try{
            // 从单元格历史列表中取出最新的单元格值
            Cell lastestCell = cellList.get(0);
            String cellValue = new String(CellUtil.cloneValue(lastestCell));
            URL url= new URL(cellValue);
            put.addColumn(
                    "cf1".getBytes(),
                    "hostname".getBytes(),
                    url.getHost().getBytes()
            );
        }
        catch (Exception ex){

            return;
        }

    }


}
